package rabbitmq

import (
	"errors"
	"fmt"
	"gitee.com/shijun123456/kcgin"
	"github.com/streadway/amqp"
	"time"
)

var Rabbitmq *rabbitMq

// 定义结构体
type rabbitMq struct {
	connect *amqp.Connection
	channel *amqp.Channel
}

// 初始化
func init() {
	url := fmt.Sprintf("amqp://%s:%s@%s:%s/%s", kcgin.AppConfig.String("rabbitmq.user"), kcgin.AppConfig.String("rabbitmq.pwd"),
		kcgin.AppConfig.String("rabbitmq.host"), kcgin.AppConfig.String("rabbitmq.port"), kcgin.AppConfig.String("rabbitmq.vhost"))
	connect, err := amqp.Dial(url)
	if err != nil {
		panic(err)
	}
	Rabbitmq = &rabbitMq{
		connect: connect,
	}
}

// 生产者
// @param  string exchange 交换机名称
// @param  string content  队列内容
// @param  string params   可变参数 第一个参数，用于赋值给routingkey 第二个参数延时队列毫秒数
// @return error
func (r *rabbitMq) Publish(exchange string, content string, params ...string) error {
	var routingKey string
	Headers := make(map[string]interface{})
	if len(params) != 0 {
		routingKey = params[0]
	}
	if len(params) > 1 {
		Headers["x-delay"] = params[1]
	}
	ch, err := r.connect.Channel()
	if err != nil {
		return errors.New(err.Error())
	}
	defer ch.Close()
	err = ch.Publish(
		exchange,   // exchange fanout
		routingKey, // routing key
		false,      // mandatory
		false,      // immediate
		amqp.Publishing{
			DeliveryMode: 2,
			ContentType:  "text/plain",
			Body:         []byte(content),
			Timestamp:    time.Now(),

			Headers: Headers,
		})
	return err
}

// 消费者
// @param string    queueName 队列名称
// @param func      消费处理
// @param interface params    可变参数，若传值、则只传一个bool值即可，用于是否收到消息需要确认，默认是需要确认
// @return <-chan amqp.Delivery
func (r *rabbitMq) Consume(queueName string, consume func(msg amqp.Delivery), params ...interface{}) error {
	ch, err := r.connect.Channel()
	if err != nil {
		return err
	}
	defer ch.Close()
	autoAck := false
	if len(params) != 0 {
		autoAck = params[0].(bool)
	}
	msgs, err := ch.Consume(
		queueName, // queue
		"",        // consumer
		autoAck,   // auto-ack
		false,     // exclusive
		false,     // no-local
		false,     // no-wait
		nil,       // args
	)
	if err != nil {
		return err
	}

	for msg := range msgs {
		consume(msg)
		err := msg.Ack(false)
		if err != nil {
			println(err)
		}
	}
	return nil
}

func (r *rabbitMq) Close() error {
	err := r.connect.Close()
	if err != nil {
		fmt.Println("close channel fail,", err.Error())
		return err
	}
	return nil
}
